package com.shiku.imserver.repository;

import cn.xyz.commons.utils.ReqUtil;
import cn.xyz.mianshi.utils.SKBeanUtils;
import cn.xyz.mianshi.vo.Room;
import com.alibaba.fastjson.JSON;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.shiku.imserver.CoreService;
import com.shiku.imserver.common.IService;
import com.shiku.imserver.common.ImConfig;
import com.shiku.imserver.common.constant.KConstants;
import com.shiku.imserver.common.message.AbstractBaseMessage;
import com.shiku.imserver.common.message.ChatMessage;
import com.shiku.imserver.common.message.PullBatchGroupMessage;
import com.shiku.imserver.common.message.PullBatchGroupRespMessage;
import com.shiku.imserver.common.message.SuccessMessage;
import com.shiku.imserver.common.packets.ImPacket;
import com.shiku.imserver.common.proto.MessageProBuf;
import com.shiku.imserver.common.utils.ProBufUtils;
import com.shiku.imserver.message.MessageFactory;
import com.shiku.imserver.message.hander.SendMessageHandler;
import com.shiku.imserver.repository.runnable.ChatMessageDBRunnable;
import com.shiku.imserver.repository.runnable.GroupMessageDBRunnable;
import com.shiku.imserver.repository.runnable.LastMessageDBRunnable;
import com.shiku.imserver.service.IMBeanUtils;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import org.bson.Document;
import org.bson.types.ObjectId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
@Slf4j
public class MessageRepository extends BaseRepository implements IService {
    private static final int GROUPOFFMESSAGE_LIMIT = 100;
    public static final String LASTCHAT_DB = "shiku_lastChats";
    private static final String MESSAGE_OFFLINE = "message_offline";
    private static final String MSGS_DB = "shiku_msgs";
    private static final String MUCMsg_ = "mucmsg_";

    private  static  final  String SHIKU_ROOM_MEMBER = "shiku_room_member";
    /* access modifiers changed from: private */
    public static Logger logger = LoggerFactory.getLogger(MessageRepository.class);
    private ChatMessageDBRunnable chatMessageDBRunnable;
    private MongoDatabase chatMsgDB;
    private GroupMessageDBRunnable groupMessageDBRunnable;
    private LastMessageDBRunnable lastMessageDBRunnable;
    private MongoDatabase mucdb;
    private MongoDatabase offMsgDB;
    private MongoDatabase roomMemberDB;
    Queue<ChatMessage> queue = new LinkedBlockingQueue();
    MessageRepositoryThread thread;

    /* JADX WARN: Type inference failed for: r7v9, types: [java.lang.Runnable, com.shiku.imserver.repository.runnable.ChatMessageDBRunnable], assign insn: 0x00a3: IGET  (r7v9 ? I:com.shiku.imserver.repository.runnable.ChatMessageDBRunnable) = (r9v0 'this' com.shiku.imserver.repository.MessageRepository A[THIS]) com.shiku.imserver.repository.MessageRepository.chatMessageDBRunnable com.shiku.imserver.repository.runnable.ChatMessageDBRunnable */
    /* JADX WARN: Type inference failed for: r7v11, types: [com.shiku.imserver.repository.runnable.GroupMessageDBRunnable, java.lang.Runnable], assign insn: 0x00b6: IGET  (r7v11 ? I:com.shiku.imserver.repository.runnable.GroupMessageDBRunnable) = (r9v0 'this' com.shiku.imserver.repository.MessageRepository A[THIS]) com.shiku.imserver.repository.MessageRepository.groupMessageDBRunnable com.shiku.imserver.repository.runnable.GroupMessageDBRunnable */
    /* JADX WARN: Type inference failed for: r7v12, types: [java.lang.Runnable, com.shiku.imserver.repository.runnable.LastMessageDBRunnable], assign insn: 0x00c7: IGET  (r7v12 ? I:com.shiku.imserver.repository.runnable.LastMessageDBRunnable) = (r9v0 'this' com.shiku.imserver.repository.MessageRepository A[THIS]) com.shiku.imserver.repository.MessageRepository.lastMessageDBRunnable com.shiku.imserver.repository.runnable.LastMessageDBRunnable */
    public boolean initialize() {
        ImConfig.MongoConfig mongoConfig = IMBeanUtils.getImconfig().getMongoConfig();
        MongoClient mongoClient = IMBeanUtils.getBeanManager().getMongoClient();
        try {
            new MongoClientURI(mongoConfig.getUri());
            this.chatMsgDB = mongoClient.getDatabase(MSGS_DB);
            this.roomMemberDB = mongoClient.getDatabase(SHIKU_ROOM_MEMBER);
            this.mucdb = mongoClient.getDatabase("imRoom");
            MongoCollection<Document> msgsCollection = this.chatMsgDB.getCollection(MSGS_DB);
            msgsCollection.createIndex(new BasicDBObject("sender", 1));
            msgsCollection.createIndex(new BasicDBObject("receiver", 1));
            msgsCollection.createIndex(new BasicDBObject("sender", 1).append("receiver", 1));
            msgsCollection.createIndex(new BasicDBObject("sender", 1).append("receiver", 1).append("ts", 1));
            this.offMsgDB = mongoClient.getDatabase(MESSAGE_OFFLINE);
            this.chatMessageDBRunnable = new ChatMessageDBRunnable(this.chatMsgDB);
            new Thread((Runnable) this.chatMessageDBRunnable).start();
            this.groupMessageDBRunnable = new GroupMessageDBRunnable(this.mucdb);
            new Thread((Runnable) this.groupMessageDBRunnable).start();
            this.lastMessageDBRunnable = new LastMessageDBRunnable();
            new Thread((Runnable) this.lastMessageDBRunnable).start();
            this.thread = new MessageRepositoryThread();
            this.thread.start();
            return true;
        } catch (Exception ex) {
            ex.printStackTrace();
            return false;
        }
    }

    public void archiveMessage(ChatMessage model) {
        if (1 == model.getMessageHead().getChatType()) {
            storeChatMessage(model);
        } else if (2 == model.getMessageHead().getChatType()) {
            storeGroupMessage(model);
        }
    }

    /* access modifiers changed from: private */
    public void storeChatMessage(ChatMessage model) {
        if (model.getType() != 0 && !model.isReadDel()) {
            if (26 == model.getType()) {
                IMBeanUtils.getMessageRepository().updateMsgReadStatus(Long.valueOf(model.getFromUserId()).longValue(), model.getContent());
            } else if (5 == model.getType() / 100) {
                String from = model.getMessageHead().getFrom();
                String to = model.getMessageHead().getTo();
                IMBeanUtils.getBeanManager().getNewFirendRepository().saveNewFirendRecord(CoreService.parseBareUserId(from), CoreService.parseBareUserId(to), from, model.getType(), model.getContent());
            } else if (9 == model.getType() / 100) {
            } else {
                if (120 >= model.getType() || 202 == model.getType()) {
                    this.chatMessageDBRunnable.putMessageToTask(model);
                    this.lastMessageDBRunnable.putLastChat(model);
                }
            }
        }
    }

    /* access modifiers changed from: private */
    public void storeGroupMessage(ChatMessage message) {
        if (message.getType() != 0 && 26 != message.getType()) {
            this.groupMessageDBRunnable.putMessageToTask(message);
            this.lastMessageDBRunnable.putLastChat(message);
        }
    }

    public ChatMessage loadGroupMessage(String roomJid, String messageId) {
        ChatMessage message = IMBeanUtils.getRedisService().queryGroupMessage(messageId);
        if (message != null) {
            return message;
        }
        Document document = (Document) this.mucdb.getCollection(MUCMsg_ + roomJid).find(new Document("messageId", messageId)).first();
        if (document == null) {
            return null;
        }
        ChatMessage message2 = (ChatMessage) JSON.parseObject(document.getString("message"), ChatMessage.class);
        if (message2 != null) {
            IMBeanUtils.getRedisService().saveGroupMessage(message2);
        }
        return message2;
    }

    public long loadGroupHistoryCount(String jid, long lastTime) {
        MongoCollection<Document> collection = this.mucdb.getCollection(MUCMsg_ + jid);
        Document query = new Document("room_jid", jid);
        query.append("timeSend", new Document("$gt", Long.valueOf(lastTime)));
        return collection.count(query);
    }

    public void pullBatchGroupHistoryMessage(ChannelContext channelContext, PullBatchGroupMessage pullBatchGroupMessageReq) {
        PullBatchGroupRespMessage groupHistoryMessage;
        if (pullBatchGroupMessageReq.getJidList() != null) {
                log.info("===www===");
            long endTime = pullBatchGroupMessageReq.getEndTime();
            String from = pullBatchGroupMessageReq.getMessageHead().getFrom();
            try {
                for (String str : pullBatchGroupMessageReq.getJidList()) {
                    if (channelContext.isClosed) {
                        logger.error(" channelContext isClosed1 => {} " + from);
                        return;
                    }
                    if (KConstants.isDebug) {
                        logger.info("pull message userId > {}  jid {}", from, str);
                    }
                    String[] split = str.split(",");
                    Document query = new Document("jid",split[0]);
                   // SKBeanUtils.getRoomManager().getRoomId()
                    Document document = this.mucdb.getCollection("shiku_room").find(query).first();
                    ObjectId roomId =(ObjectId)document.get("_id");

                    String collectionName = getCollectionName(roomId);

                    Document queryRoom = new Document("roomId", roomId).append("userId",Integer.valueOf(channelContext.userid));

                    Document RommMemberDocument = this.roomMemberDB.getCollection(collectionName).find(queryRoom).first();


                    long startTime =  Long.valueOf(split[1]).longValue();
                    log.info("[ww]startTime="+startTime+",collectionName="+collectionName+",channelContext.userid="+channelContext.userid+",JID="+split[0]+",roomId="+roomId+",getCounter="+ roomId.getCounter());
                    if(RommMemberDocument != null)
                    {
                        long createTime = RommMemberDocument.getLong("createTime") * 1000;
                        log.info("[ww]createTime="+createTime);
                        if (startTime > 0 &&  startTime < createTime)
                        {
                            startTime = createTime;
                        }
                    }
                    if (2 <= split.length && (groupHistoryMessage = queryGroupHistoryMessage(split[0],startTime, endTime)) != null) {
                        log.info("===ww");
                        groupHistoryMessage.setMessageId(pullBatchGroupMessageReq.getMessageHead().getMessageId());
                        ImPacket imPacket = ProBufUtils.encodeImPacket(groupHistoryMessage, MessageProBuf.PullGroupMessageRespProBuf.getDescriptor());
                        imPacket.setCommand((short)15);
                        CoreService.send(channelContext, imPacket);
                    }
                }

            } catch (Exception e) {
                logger.error(e.getMessage());
            }
        }
    }
/**
 * 进入群时候，拉取历史消息
 * @param jid
 * @param seconds
 * @return
 */
    public List<ChatMessage> loadGroupHistoryMessage(String jid, long seconds) {
        List<ChatMessage> result = new ArrayList<>();
        Document query = new Document("room_jid", jid);
        query.append("timeSend", new Document("$gt", Long.valueOf(System.currentTimeMillis() - seconds)));
        MongoCursor<Document> iterator = this.mucdb.getCollection(MUCMsg_ + jid).find(query).sort(new Document("timeSend", -1)).limit((int) GROUPOFFMESSAGE_LIMIT).iterator();
        while (iterator.hasNext()) {
            try {
                ChatMessage message = (ChatMessage) JSON.parseObject(((Document) iterator.next()).getString("message"), ChatMessage.class);
                if (message != null) {
                    result.add(message);
                }
            } catch (Exception e) {
                logger.error(" Exception {}", e.getMessage());
            } catch (Throwable th) {
                if (iterator != null) {
                    iterator.close();
                }
                throw th;
            }
        }
        if (iterator != null) {
            iterator.close();
        }
        Collections.reverse(result);
        return result;
    }

    public PullBatchGroupRespMessage queryGroupHistoryMessage(String jid, long startTime, long endTime) {
        int limit = GROUPOFFMESSAGE_LIMIT;
        Document query = new Document("room_jid", jid);

        query.append("timeSend", new Document("$gt", Long.valueOf(startTime)).append("$lt", Long.valueOf(endTime+5000)));
        MongoCollection<Document> collection = this.mucdb.getCollection(MUCMsg_ + jid);
        long count = collection.count(query);
        if (0 == count) {
            return null;
        }
        if (count < 100) {
            limit = (int) count;
        }
        MongoCursor<Document> iterator = collection.find(query).sort(new Document("timeSend", -1)).limit(limit).iterator();
        List<ChatMessage> messageList = new ArrayList<>();
        while (iterator.hasNext()) {
            try {
                ChatMessage message = (ChatMessage) JSON.parseObject(((Document) iterator.next()).getString("message"), ChatMessage.class);
                if (message != null) {
                	logger.info("==queryGroupHistoryMessage=getContent="+message.getContent());
                    messageList.add(message);
                }
            } catch (Exception e) {
                logger.error(" Exception {}", e.getMessage());
            } catch (Throwable th) {
                if (iterator != null) {
                    iterator.close();
                }
                throw th;
            }
        }
        if (iterator != null) {
            iterator.close();
        }
        Collections.reverse(messageList);
        PullBatchGroupRespMessage result = new PullBatchGroupRespMessage();
        result.setJid(jid);
        result.setCount(count);
        result.setMessageList(messageList);
        return result;
    }
    /**
     * 拉取群消息
     * @param jid
     * @param size
     * @param startTime
     * @param endTime
     * @return
     */
    public List<ChatMessage> pullGroupHistoryMessage(String jid, int size, long startTime, long endTime) {
        List<ChatMessage> result = new ArrayList<>();
        Document query = new Document("room_jid", jid);
        MongoCollection<Document> collection = this.mucdb.getCollection(MUCMsg_ + jid);
        if (0 < startTime) {
            query.append("timeSend", new Document("$gt", Long.valueOf(startTime)));
        }
        if (0 < endTime) {
            query.append("timeSend", new Document("$lt", Long.valueOf(endTime)));
        }
        MongoCursor<Document> iterator = collection.find(query).sort(new Document("timeSend", -1)).limit(size).iterator();
        while (iterator.hasNext()) {
            try {
                ChatMessage message = (ChatMessage) JSON.parseObject(((Document) iterator.next()).getString("message"), ChatMessage.class);
                if (message != null) {
                    result.add(message);
                }
            } catch (Exception e) {
                logger.error(" Exception {}", e.getMessage());
            } catch (Throwable th) {
                if (iterator != null) {
                    iterator.close();
                }
                throw th;
            }
        }
        if (iterator != null) {
            iterator.close();
        }
        Collections.reverse(result);
        return result;
    }
    /**
     * 单聊消息修改为已读
     * @param sender
     * @param msgId
     */
    public void updateMsgReadStatus(long sender, String msgId) {
        if (msgId != null) {
            getCollection(this.chatMsgDB, sender).updateOne(new Document("messageId", msgId), new Document("$set", new BasicDBObject("isRead", 1)));
        }
    }

    public void deleteChatOffToJID(String userJid) {
    	
    }
    public void deleteChatOffToMessageID(String msgID,String userJid)
    {
    	 Document query = new Document("messageId", msgID);
    	 MongoCollection<Document> collection = getCollection(this.offMsgDB, (long) Integer.valueOf(userJid).intValue());
    	 //从离线消息数据库，删除收到状态为已读消息
    	 collection.deleteMany(query);
    }
    public ChatMessage loadChatMessage(String userId, String messageId) {
        ChatMessage message = IMBeanUtils.getRedisService().queryChatMessage(messageId);
        if (message != null) {
            return message;
        }
        Document document = (Document) getCollection(this.chatMsgDB, (long) Integer.valueOf(userId).intValue()).find(new Document("messageId", messageId)).first();
        if (document == null) {
            return null;
        }
        ChatMessage message2 = (ChatMessage) JSON.parseObject(document.getString("message"), ChatMessage.class);
        if (message2 != null) {
            IMBeanUtils.getRedisService().saveChatMessage(message2);
        }
        return message2;
    }
/**
 * 拉取单聊的历史消息
 * @param userJid
 * @param toJid
 * @param size
 * @param startTime
 * @param endTime
 * @return
 */
    public List<ChatMessage> pullChatHistoryMessage(String userJid, String toJid, int size, long startTime, long endTime) {
        List<ChatMessage> result = new ArrayList<>();
        Document query = new Document("sender", userJid);
        query.append("receiver", toJid);
        if (0 < startTime) {
            query.append("timeSend", new Document("$gt", Long.valueOf(startTime)));
        }
        if (0 < endTime) {
            query.append("timeSend", new Document("$lt", Long.valueOf(endTime)));
        }
        MongoCursor<Document> iterator = getCollection(this.chatMsgDB, (long) Integer.valueOf(userJid).intValue()).find(query).sort(new Document("timeSend", -1)).limit(size).iterator();
        while (iterator.hasNext()) {
            try {
                ChatMessage message = (ChatMessage) JSON.parseObject(((Document) iterator.next()).getString("message"), ChatMessage.class);
                if (message != null) {
                    result.add(message);
                }
            } catch (Exception e) {
                logger.error(" Exception {}", e.getMessage());
            }
        }
        Collections.reverse(result);
        return result;
    }
/**
 * 用户上线后，加载离线的单聊消息
 * @param userJid
 * @param delete
 * @return
 */
    public List<ChatMessage> loadChatOffToJID(String userJid, boolean delete) {
    	log.info("===loadChatOffToJID=userJid="+userJid);
        List<ChatMessage> result = new LinkedList<>();
        Document query = new Document("receiver", userJid);
        Document sort = new Document("timeSend", 1);
        MongoCollection<Document> collection = getCollection(this.offMsgDB, (long) Integer.valueOf(userJid).intValue());
        MongoCursor<Document> iterator = collection.find(query).sort(sort).iterator();
        while (iterator.hasNext()) {
            try {
                ChatMessage message = (ChatMessage) JSON.parseObject(((Document) iterator.next()).getString("message"), ChatMessage.class);
                if (message != null) {
                    result.add(message);
                }
            } catch (Exception e) {
                logger.error(" Exception {}", e.getMessage());
            } catch (Throwable th) {
                if (iterator != null) {
                    iterator.close();
                }
                throw th;
            }
        }
        if (iterator != null) {
            iterator.close();
        }
        //删除单聊记录消息  离线重回可能马上又断线，导致消息发送失败，需要在收到回执以后再来删除消息
        //collection.deleteMany(query);
        return result;
    }
/**
 * 单聊，对方不在线，保存聊天消息
 * @param message
 */
    public void storeChatOffMessage(ChatMessage message) {
        long userId;
        try {
            Document dbObj = new Document();
            dbObj.put("message", message.toString());
            dbObj.put("to", message.getMessageHead().getTo());
            dbObj.put("from", message.getMessageHead().getFrom());
            dbObj.put("sender", message.getFromUserId());
            dbObj.put("receiver", message.getToUserId());
            dbObj.put("contentType", Short.valueOf(message.getType()));
            dbObj.put("messageId", message.getMessageHead().getMessageId());
            dbObj.put("timeSend", Long.valueOf(message.getTimeSend()));
            dbObj.put("content", message.getContent());
            try {
                userId = Long.valueOf(message.getToUserId()).longValue();
            } catch (NumberFormatException e) {
                userId = Long.valueOf(CoreService.parseBareUserId(message.getMessageHead().getTo())).longValue();
            }
            getCollection(this.offMsgDB, userId).insertOne(dbObj);
        } catch (Exception e2) {
            logger.error(e2.getMessage(), e2);
        }
    }
    public boolean getChatOffMessageAddFirend(String fromId,String receiverId)
    {
         Document query = new Document("receiver", receiverId);
         query.append("sender",fromId);
         query.append("contentType",500);
         MongoCollection<Document> collection = getCollection(this.offMsgDB, (long) Integer.valueOf(receiverId).intValue());
         MongoCursor<Document> iterator = collection.find(query).iterator();
         if(iterator.hasNext()) return true;
         return false;
    }
    /**
     * 聊天消息，插入队列等待保存到数据库
     * @param message
     * @return
     */
    public boolean offerMessage(ChatMessage message) {
        return this.queue.offer(message);
    }
    /**
     * 获取群聊天，最后一条id
     * @param jid
     * @return
     */
    public long getGroupMaxID(String jid)
    {	
    	MongoCollection<Document> collection = this.mucdb.getCollection(MUCMsg_ + jid);
    	MongoCursor<Document> iterator = collection.find().sort(new Document("chatMsgId", -1)).limit(1).iterator();
    	if(iterator.hasNext())
    	{
    		try {
    			long chatid = ((Document)iterator.next()).getLong("chatMsgId");
    			return chatid; 
    		}
    		catch(Exception e)
    		{
    			return 0;
    		}
    		
    		
    	}
    	// MongoCursor<Document> iterator = collection.find(query).sort(new Document("timeSend", -1)).limit(size).iterator();
    	return 0;
    }
/**
 * 聊天数据入库
 * @author wuwenbin
 *
 */
    public class MessageRepositoryThread extends Thread {
        public MessageRepositoryThread() {
        }

        public void run() {
            while (true) {
                try {
                    if (MessageRepository.this.queue.isEmpty()) {
                        try {
                            //Thread.sleep(2000);
                            Thread.sleep(100);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    } else {
                        ChatMessage message = MessageRepository.this.queue.poll();
                        if (1 == message.getMessageHead().getChatType()) {
                            MessageRepository.this.storeChatMessage(message);
                        } else if (2 == message.getMessageHead().getChatType()) {
                            MessageRepository.this.storeGroupMessage(message);
                        }
                    }
                } catch (Exception e2) {
                    MessageRepository.logger.error(e2.getMessage());
                }
            }
        }
    }
}
